package com.goldman.aplt.service.mq.impl;

import com.goldman.aplt.base.constant.ApConstants;
import com.goldman.aplt.base.type.BaseEnumType;
import com.goldman.aplt.base.type.MsgEnumType;
import com.goldman.aplt.entity.mq.ApbMqProducerMsg;
import com.goldman.aplt.mq.model.MsgJoinMkActiveDto;
import com.goldman.aplt.mq.producer.MkMessagePublisher;
import com.goldman.aplt.service.mq.ApbMqProducerMsgService;
import com.goldman.aplt.service.mq.MessageProducer;
import com.goldman.aplt.util.ApRedisDistributeLock;
import com.goldman.aplt.util.BeanContext;
import com.goldman.aplt.util.CommUtil;
import com.goldman.message.common.event.MessageModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.*;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class MessageProducerImpl implements MessageProducer {

    @Autowired
    private ApbMqProducerMsgService apbMqProducerMsgService;

    @Autowired
    private ApRedisDistributeLock apRedisDistributeLock;

    @Override
    public void saveMessage(MessageModel messageModel) {
        byte[] byteArray = new byte[1];
        try {
            ByteArrayOutputStream fileOutputStream = new ByteArrayOutputStream();
            ObjectOutputStream objectOutputStream = new ObjectOutputStream(fileOutputStream);
            objectOutputStream.writeObject(messageModel);
            byteArray = fileOutputStream.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        String msgId = messageModel.getMsgId();
        if (CommUtil.isNull(msgId)) {
            msgId = getUUID();
        }
        String type = messageModel.getType();
        ApbMqProducerMsg apbMqProducerMsg = ApbMqProducerMsg.builder()
                .messageId(msgId)
                .message(byteArray)
                .status(MsgEnumType.E_MQ_MSG_STATUS.TODO)
                .messageType(type)
                .destination(messageModel.getDestination())
                .build();
        apbMqProducerMsgService.save(apbMqProducerMsg);
    }

    public void updateMsgStatusSend(String msgId, MsgEnumType.E_MQ_MSG_STATUS status) {
        ApbMqProducerMsg msgServiceById = apbMqProducerMsgService.getById(msgId);
        msgServiceById.setStatus(status);
        apbMqProducerMsgService.updateById(msgServiceById);
    }

    @Override
    public Object getMessage(String msgId) {
        ApbMqProducerMsg one = apbMqProducerMsgService.lambdaQuery().eq(ApbMqProducerMsg::getMessageId, msgId).one();
        byte[] message = one.getMessage();
        Object obj = null;
        try {
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(message);
            ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
            obj = objectInputStream.readObject();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
        return obj;
    }

    public static String getUUID() {
        ThreadLocalRandom random = ThreadLocalRandom.current();
        String uuid = (new UUID((long)random.nextInt(), (long)random.nextInt())).toString();
        return uuid.replaceAll("-", "");
    }

    /**
     *  message定时发送
     */
    public void autoSendMq(){
        List<ApbMqProducerMsg> list = apbMqProducerMsgService.lambdaQuery().eq(ApbMqProducerMsg::getStatus, "1").list();
        for (ApbMqProducerMsg  apbMqProducerMsg : list) {
            String messageId = apbMqProducerMsg.getMessageId();

            String key = ApConstants.MKB_RDS_lOCK_PREFIX + BaseEnumType.LOCK_EVENT.MQ_MESSAGE_SEND + messageId;

            boolean isLocked = apRedisDistributeLock.tryLock(key, 1000, 60 * 1000, TimeUnit.MILLISECONDS);
            if (!isLocked) {
                log.info("MqMessageSendJob--id：{}数据被锁定", messageId);
                return;
            }
            try {
                String messageType = apbMqProducerMsg.getDestination();
                if(MsgEnumType.E_TOPIC.MARKET_TOPIC.getValue().equals(messageType)){
                    MessageModel message = (MsgJoinMkActiveDto)getMessage(messageId);
                    MkMessagePublisher mkActivePublisher = BeanContext.getBean(MkMessagePublisher.class);
                    boolean publish = mkActivePublisher.publish(message);
                    if(publish){
                        updateMsgStatusSend(messageId, MsgEnumType.E_MQ_MSG_STATUS.DONE);
                    }else{
                        updateMsgStatusSend(messageId, MsgEnumType.E_MQ_MSG_STATUS.ERR);
                    }
                }
            } catch (BeansException e) {
                log.info("MqMessageSendJob--id：{}发送失败", messageId);
                return;
            } finally {
                apRedisDistributeLock.unlock(key);
            }
        }
    }
}
